use std::{
    fmt::{Display, Formatter},
    future::Future,
    io, mem,
    pin::Pin,
    task::{Context, Poll},
    thread,
};

use crate::{
    accept::Accept,
    builder,
    handle::ServerHandle,
    service::InternalServiceFactory,
    signals::{SignalKind, Signals},
    waker_queue::{WakerInterest, WakerQueue},
    worker::{ServerWorkerConfig, WorkerHandleServer},
    ServerBuilder,
};
use futures_core::{future::BoxFuture, Stream};

// 对futures扩展
use futures_util::stream::StreamExt as _;

use tokio::{
    signal,
    sync::{mpsc::UnboundedReceiver, oneshot},
};
pub(crate) enum ServerCommand {
    WorkerFaulted(usize),
    Pause(oneshot::Sender<()>),
    Resume(oneshot::Sender<()>),
    Stop {
        graceful: bool,
        completion: Option<oneshot::Sender<()>>,
        force_system_stop: bool,
    },
}

pub struct Server {
    handle: ServerHandle,
    fut: BoxFuture<'static, io::Result<()>>,
}

impl Server {
    pub fn build() -> ServerBuilder {
        ServerBuilder::default()
    }

    pub(crate) fn new(builder: ServerBuilder) -> Self {
        Self {
            handle: ServerHandle::new(builder.cmd_tx.clone()),
            fut: Box::pin(ServerInner::run(builder)),
        }
    }
}

impl Future for Server {
    type Output = io::Result<()>;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        Pin::new(&mut Pin::into_inner(self)).poll(cx)
    }
}

pub struct ServerInner {
    worker_handles: Vec<WorkerHandleServer>,
    accept_handle: Option<thread::JoinHandle<()>>,
    worker_config: ServerWorkerConfig,
    services: Vec<Box<dyn InternalServiceFactory>>,
    waker_queue: WakerQueue,
    stopping: bool,
    system_stop: bool,
}

impl ServerInner {
    async fn run(builder: ServerBuilder) -> io::Result<()> {
        let (mut this, mut mux) = Self::run_sync(builder)?;
        while let Some(cmd) = mux.next().await {
            this.handle_cmd(cmd).await;
            if this.stopping {
                break;
            }
        }

        Ok(())
    }

    fn run_sync(mut builder: ServerBuilder) -> io::Result<(Self, ServerEventMultiplexer)> {
        //TODO: 日志打印
        let sockets = mem::take(&mut builder.sockets)
            .into_iter()
            .map(|t| (t.0, t.2))
            .collect();
        // Accept.start
        let (waker_queue, worker_handles, accept_handle) = Accept::start(sockets, &builder)?;

        let mux = ServerEventMultiplexer {
            signal_fut: (builder.listen_os_signals).then(Signals::new),
            cmd_rx: builder.cmd_rx,
        };

        let server = ServerInner {
            waker_queue,
            accept_handle: Some(accept_handle),
            worker_handles,
            worker_config: builder.worker_config,
            services: builder.factories,
            system_stop: builder.exit,
            stopping: false,
        };

        Ok((server, mux))
    }

    /// 处理服务指令
    async fn handle_cmd(&mut self, item: ServerCommand) {
        match item {
            ServerCommand::WorkerFaulted(_) => todo!(),
            ServerCommand::Pause(sender) => {
                self.waker_queue.wake(WakerInterest::Pause);
                sender.send(());
            }
            ServerCommand::Resume(sender) => {
                self.waker_queue.wake(WakerInterest::Resume);
                sender.send(());
            }
            ServerCommand::Stop {
                graceful,
                completion,
                force_system_stop,
            } => todo!(),
        }
    }

    fn map_signal(signal: SignalKind) -> ServerCommand {

        //TODO:

        match signal {
            SignalKind::Int => {
                println!("SIGINT received; starting forced shutdown");
                ServerCommand::Stop {
                    graceful: false,
                    completion: None,
                    force_system_stop: true,
                }
            }
            SignalKind::Term => todo!(),
            SignalKind::Quit => todo!(),
        }
    }
}

// 多路复用
struct ServerEventMultiplexer {
    cmd_rx: UnboundedReceiver<ServerCommand>,
    signal_fut: Option<Signals>,
}

impl Stream for ServerEventMultiplexer {
    type Item = ServerCommand;
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        // 取出内部值
        let this = Pin::into_inner(self);
        if let Some(signal_fut) = &mut this.signal_fut {
            if let Poll::Ready(signal) = Pin::new(signal_fut).poll(cx) {
                this.signal_fut = None;
                return Poll::Ready(Some(ServerInner::map_signal(signal)));
            }
        }
        this.cmd_rx.poll_recv(cx)
    }
}
